package org.fujene.communicate.filter;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.TreeSet;

public class SimpleFilter implements Filter<long[], long[]>{
	@Override
	public long[] intersection(long[] dummy, long[]... part){
		ArrayList<Long> results = new ArrayList<Long>();

		int parts = part.length;
		long[] currentID = new long[parts];
		int[] currentPoint = new int[parts];
		for(int i = 0; i < parts; i++){
			currentID[i] = part[i][0];
			currentPoint[i] = 0;
		}

		boolean isSame, isBreak;
		long firstNum;
		int firstIndex;
		while(true){
			isSame = true;
			firstNum = Long.MAX_VALUE;
			firstIndex = 0;
			for(int i = 0; i < parts; i++){
				if(currentID[i] < firstNum){
					firstIndex = i;
					firstNum = currentID[i];
				}
				if(i == parts - 1) break;
				if(currentID[i] != currentID[i + 1]) isSame = false;
			}

			isBreak = false;
			if(isSame){
				// Same pointer is found
				results.add(Long.valueOf(currentID[0]));
				for(int i = 0; i < parts; i++){
					if(currentPoint[i] == part[i].length - 1){
						isBreak = true;
						break;
					}
					currentPoint[i]++;
					currentID[i] = part[i][currentPoint[i]];
				}
			}else{
				if(currentPoint[firstIndex] == part[firstIndex].length - 1){
					isBreak = true;
					break;
				}
				currentPoint[firstIndex]++;
				currentID[firstIndex] = part[firstIndex][currentPoint[firstIndex]];
			}
			if(isBreak) break;
		}

		long[] returns = new long[results.size()];
		for(int i = 0; i < results.size(); i++)
			returns[i] = results.get(i).longValue();

		return returns;
	}

	@Override
	public long[] join(long[] dummy, long[]... part){
		TreeSet<Long> results = new TreeSet<Long>();

		int parts = part.length;
		long[] currentID = new long[parts];
		BitSet reachEnd = new BitSet(parts);
		for(int i = 0; i < parts; i++)
			currentID[i] = part[i][0];

		// Stop if all parts get end point
		int count = 0;
		while(reachEnd.cardinality() != parts){
			for(int i = 0; i < parts; i++){
				if(!reachEnd.get(i)) results.add(part[i][count]);
				if(count == part[i].length - 1) reachEnd.set(i);
			}
			count++;
		}
		
		long[] returns = new long[results.size()];
		count = 0;
		for(Long value: results)
			returns[count++] = value.longValue();
				
		return returns;
	}

	@Override
	public long[] filter(long[] dummy, long[] list){
		return list;
	}
}
